package org.pizza.rocket.core.consumer;

import com.aliyun.openservices.ons.api.Message;
import org.pizza.rocket.core.annotation.RocketMessageListener;
import org.pizza.rocket.core.listener.BatchRocketMQListener;
import org.pizza.rocket.core.listener.OrderRocketMQListener;
import org.pizza.rocket.core.listener.SimpleRocketMQListener;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.aop.framework.AopProxyUtils;
import org.springframework.core.env.StandardEnvironment;

import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.nio.charset.Charset;
import java.util.Objects;

/**
 * @author kynb_club
 * email kynb_club@163.com
 * @since 2020/11/24 11:33 上午
 */
@Slf4j
public abstract class AbstractConsumerWrapper implements ConsumerWrapper{
    /**
     * 环境
     */
    protected StandardEnvironment environment;
    /**
     * 是否运行状态
     */
    protected boolean running;
    /**
     * 是否运行状态
     */
    protected RocketMessageListener annotation;
    /**
     * 是否运行状态
     */
    protected static final String CHARSET = "UTF-8";
    /**
     * 泛型的类型
     */
    protected Class messageType;

    protected ObjectMapper objectMapper;
    protected Class getMessageType(Object obj) {
        Class<?> targetClass = AopProxyUtils.ultimateTargetClass(obj);
        Type[] interfaces = targetClass.getGenericInterfaces();
        Class<?> superclass = targetClass.getSuperclass();
        while (0 == interfaces.length && Objects.nonNull(superclass)) {
            interfaces = superclass.getGenericInterfaces();
            superclass = targetClass.getSuperclass();
        }
        for (Type type : interfaces) {
            if (type instanceof ParameterizedType) {
                ParameterizedType parameterizedType = (ParameterizedType) type;
                if (Objects.equals(parameterizedType.getRawType(), SimpleRocketMQListener.class)
                    || Objects.equals(parameterizedType.getRawType(), OrderRocketMQListener.class)
                    || Objects.equals(parameterizedType.getRawType(), BatchRocketMQListener.class)
                ) {
                    Type[] actualTypeArguments = parameterizedType.getActualTypeArguments();
                    if (Objects.nonNull(actualTypeArguments) && actualTypeArguments.length > 0) {
                        return (Class) actualTypeArguments[0];
                    } else {
                        return Object.class;
                    }
                }
            }
        }
        return Object.class;
    }
    protected Object doConvertMessage(Message message) {
        if (Objects.equals(messageType, Message.class)) {
            return message;
        } else {
            String str = new String(message.getBody(), Charset.forName(CHARSET));
            if (Objects.equals(messageType, String.class)) {
                return str;
            } else {
                // If msgType not string, use objectMapper change it.
                try {
                    return objectMapper.readValue(str, messageType);
                } catch (Exception e) {
                    log.info("convert failed. str:{}, msgType:{}", str, messageType);
                    throw new RuntimeException("cannot convert message to " + messageType, e);
                }
            }
        }
    }
    public void setRunning(boolean running) {
        this.running = running;
    }

    public StandardEnvironment getEnvironment() {
        return environment;
    }

    public void setEnvironment(StandardEnvironment environment) {
        this.environment = environment;
    }

    public ObjectMapper getObjectMapper() {
        return objectMapper;
    }

    public void setObjectMapper(ObjectMapper objectMapper) {
        this.objectMapper = objectMapper;
    }
}
